#!/home/aaron/pyenv/bin/python
"""
execute single supply task by moia
"""
from decimal import Decimal
import sys, os
from pathlib import Path
from transferfile import TransferFactory

BASE_DIR = Path(__file__).resolve().parent
os.chdir(str(BASE_DIR))
print("current_path = ", Path.cwd())
from somedecorators import timeit, retry, email_on_exception
from dbinterface.database_client import DataBaseClientFactory
from log import get_logger
from datetime import date, datetime, timedelta
import subprocess
import utils


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
logger = get_logger("/home/hyshf/HRDS/hdptrans/aaron/file/logs", "supply.py.log", 100)


class ShellErrorException(Exception):
    pass


class SupplyTask(object):
    def __init__(self, task_no: int, biz_dt: date, isLast: bool, test_mode=False):
        self.max_retry = 3
        self.data_num = 0
        self.task_no = task_no
        self.delimter = ","
        self.sys_no = None

        self.config_dbcli = DataBaseClientFactory.create(
            dbtype="mysql",
            host="192.168.172.133",
            port=3306,
            user="root",
            pwd="Mysql@96068",
            database="damx",
        )
        # self.config_dbcli.set_current_schema("damx")
        self.isLast = isLast
        for task in self.config_dbcli.read_map(
            "select * from damx.t_subscribe_task where tsk_no = %s",
            (task_no,)
            # "select * from damx.t_subscribe_system where biz_dt = %s",(datetime(2020,8,5),)
        ):
            self.tsk_nm = task["tsk_nm"]  # 任务名称';
            self.sys_no = task["sys_no"]  # 系统编号';
            logger.info(f"任务编号 = {task_no} 任务名称 = {self.tsk_nm}")
            self.prd = task["prd"]  # 周期 0 按天 1 按周 2 按旬 3 按月 4 按季 5 半年 6 按年';
            self.prd_dt = task["prd_dt"]  # 周期日期：表示周期第一天之后的第n天'

            self.quote_flg = (
                "" if task["quote_flg"] is None else task["quote_flg"]
            )  # 引号标志';
            if self.quote_flg.lower().startswith("0x"):
                self.quote_flg = chr(int(self.quote_flg, 16))
            self.delimeter = task["delimeter"]  # 分隔符';
            if self.delimeter.lower().startswith("0x"):
                self.delimeter = chr(int(self.delimeter, 16))

            ##是否所有字段均作为字符串处理，默认是
            self.all_str = task["all_str"]

            ## 做参数替换
            self.sql_text = utils.replace_yyyymmdd(task["sql_text"], biz_dt)  # sql';
            self.file_nm = utils.replace_yyyymmdd(task["file_nm"], biz_dt)
            self.cd_mod = task["cd_mod"] or "utf-8"
            self.cd_mod = self.cd_mod.lower()
            self.upload_ok_file = task["upload_ok_file"]
            logger.info(
                f"文件名 = {self.file_nm}，编码 = {self.cd_mod} 引号 = 「{task['quote_flg']}」，分割符 = 「{task['delimeter']}」"
            )

            self.vld_flg = task["vld_flg"]  # 有效标志';
            self.oprt_st = task["oprt_st"]  # 运行状态[0已完成 1运行中]';
            self.data_src_no = task["data_src_no"]  # 运行状态[0已完成 1运行中]';
            break

        if not self.sys_no:
            print(f"task {task_no} not found ")
            exit(-1)

        for cfg in self.config_dbcli.read_map(
            "select * from damx.t_subscribe_system where sys_no = %s", (self.sys_no,)
        ):
            self.sys_nm = cfg["sys_nm"]
            self.ip_addr = cfg["ip_addr"]
            logger.info(
                f"系统编号 = {self.sys_no} 系统名称 = {self.sys_nm}，目标IP = {self.ip_addr}"
            )
            self.usr_nm = cfg["usr_nm"]
            self.psw = cfg["psw"]
            self.prvd_data_mod = cfg["prvd_data_mod"]
            self.lcl_rte = utils.replace_yyyymmdd(cfg["lcl_rte"], biz_dt)
            if os.path.exists(self.lcl_rte):
                pass
            else:
                os.makedirs(self.lcl_rte)

            self.trgt_rte = utils.replace_yyyymmdd(cfg["trgt_rte"], biz_dt)
            # self.biz_dt = cfg['BIZ_DT']
            self.biz_dt = biz_dt
            logger.info(f"业务日期 = {self.biz_dt}")
            self.oprt_st = cfg["oprt_st"]
            self.upl_flg = cfg["upl_flg"]
            self.cfg_vld_flg = cfg["vld_flg"]
            if self.cfg_vld_flg == 0 and not test_mode:  # 无效
                logger.info(f"sys_nm = {self.sys_nm} is invalid, exit.")
                exit(0)
            break

        for db in self.config_dbcli.read_map(
            "select * from damx.t_subscribe_datasource where data_src_no = %s",
            (self.data_src_no,),
        ):
            self.data_src_nm = db["data_src_nm"]
            self.dtbs_tp = db["dtbs_tp"]
            self.db_ip_addr = db["ip_addr"]
            self.pt_no = db["pt_no"]
            self.dtbs_nm = db["dtbs_nm"]
            self.db_usr_nm = db["usr_nm"]
            self.db_pwd = db["psw"]
            break

        # config_dbcli.close()
        # can not close,because record log

        if self.dtbs_tp == "postgresql":
            self.dtbs_tp = "postgres"
            local_ip = utils.get_local_ip()
            if local_ip == self.db_ip_addr:
                self.db_ip_addr = "localhost"

        self.dbcli = DataBaseClientFactory.create(
            dbtype=self.dtbs_tp,
            host=self.db_ip_addr,
            port=self.pt_no,
            user=self.db_usr_nm,
            pwd=self.db_pwd,
            database=self.dtbs_nm,
        )

        ## files
        self.data_file = Path(self.lcl_rte) / self.file_nm
        self.remote_data_file = Path(self.trgt_rte) / self.file_nm

        self.ok_file = Path(self.lcl_rte) / f"{self.file_nm}.ok"
        self.remote_ok_file = Path(self.trgt_rte) / f"{self.file_nm}.ok"

        self.finish_flag = Path(self.lcl_rte) / "finish.flag"
        self.remote_finish_flag = Path(self.trgt_rte) / "finish.flag"

        # db_client
        self.dbcli = None
        if self.dtbs_tp == "postgresql":
            self.dtbs_tp = "postgres"
            local_ip = utils.get_local_ip()
            if local_ip == self.db_ip_addr:
                self.db_ip_addr = "localhost"

        # file transfer client
        self.transfer_cli = None


    def record_tsk_begin(self, tsk_no):
        sql = """
        REPLACE INTO damx.t_subscribe_task_log(TSK_NO, BIZ_DATE, LOG_LV, BGN_TM,  OPRT_ST) VALUES(%s,%s,'INFO',CURRENT_TIMESTAMP(),1)
        """
        self.config_dbcli.write(sql, (tsk_no, self.biz_dt))

    def record_tsk_end(self, tsk_no, data_num):
        sql = "update damx.t_subscribe_task_log set END_TM = current_timestamp(), DATA_NUM = %s ,OPRT_ST = 0 where tsk_no = %s and BIZ_DATE = %s "
        self.config_dbcli.write(sql, (data_num, tsk_no, self.biz_dt))

    @timeit(logger)
    def copy_to_file(self):

        if self.dtbs_tp != "postgres":
            raise Exception("非 PostGreSQL 数据库不支持 copy to file")
        if self.quote_flg:
            raise Exception("PostGreSQL 的 copy to 不支持字段增加引号标志")
        logger.info("using PostGreSQL copy to file")
        utils.chmod777(self.data_file.parent.as_posix())
        self.data_file.touch(0o777)
        utils.chmod777(self.data_file.as_posix())
        row_count = 0
        if self.cd_mod != "utf-8":
            tmp = self.data_file.with_name(self.data_file.name + ".tmp")
            row_count = self.dbcli.copy_to_file(
                self.sql_text, tmp.as_posix(), delimiter=self.delimeter
            )
            utils.chmod777(tmp.as_posix())
            r = utils.convert_encoding(
                tmp.as_posix(), self.data_file.as_posix(), "utf-8", self.cd_mod
            )
            if r == 0:
                tmp.unlink()
            else:
                logger.error(
                    f"iconv -f utf-8 -t {self.cd_mod} -c {tmp.as_posix()} -o {p.as_posix()} 失败，请查明原因"
                )
                exit(-1)
        else:
            row_count = self.dbcli.copy_to_file(
                self.sql_text, self.data_file.as_posix(), delimiter=self.delimeter
            )
        utils.chmod777(self.data_file.as_posix())
        return row_count

    @timeit(logger)
    def export(self):
        if not self.dbcli:
            print(self.db_usr_nm,self.db_pwd)
            self.set_dbcli(
                type=self.dtbs_tp,
                host=self.db_ip_addr,
                port=self.pt_no,
                user=self.db_usr_nm,
                pwd=self.db_pwd,
                database=self.dtbs_nm,
            )
        logger.info(f"sql_text:\n\n {self.sql_text} \n")
        logger.info(f"begin export data to \n\n{self.data_file}\n")
        if self.dtbs_tp == "postgres" and not self.quote_flg:
            row_counter = self.copy_to_file()
        else:
            row_counter = self.dbcli.export(
                sql=self.sql_text,
                params=(),
                file_path=self.data_file.as_posix(),
                encoding=self.cd_mod,
                delimeter=self.delimeter,
                quote=self.quote_flg,
                all_col_as_str=self.all_str,
            )
        self.ok_file.write_text(
            f"rows={row_counter}\nThis file indicates that the export of {self.data_file} has completed.\n"
        )

        self.dbcli.close()
        self.data_num = row_counter
        logger.info(f"export {row_counter} rows")
        if not self.data_file.exists():
            raise Exception(f"{self.data_file} 文件不存在，请查明原因")
        return self.data_num

    def touch_finish_flag(self):
        self.finish_flag.write_text(
            f"This file indicates that all transfer to {self.trgt_rte} have completed.\n"
        )

    @timeit(logger)
    @retry(
        times=3,
        wait_seconds=30,
        traced_exceptions=(
            ShellErrorException,
            subprocess.TimeoutExpired,
            subprocess.CalledProcessError,
        ),
    )
    def upload(self, local_file, remote_file):
        if not Path(local_file).exists():
            raise FileNotFoundError(f"{file_path} not found!")
        logger.info(
            f"begin upload \n{local_file}\n to:\n{self.ip_addr}: {remote_file}\n"
        )
        self.transfer_cli.put(local_file, remote_file)

    def get_transfer_cli(self, type, host, username, password, encoding):
        return TransferFactory.create(type, host, username, password, encoding=encoding)

    def set_dbcli(self, type, host, port, user, pwd, database):
        self.dtbs_tp = type
        self.dbcli = DataBaseClientFactory.create(
            dbtype=type, host=host, port=port, user=user, pwd=pwd, database=database
        )

    def need_execute(self) -> bool:

        # 如果文件已经存在，可以不需要卸数
        # to-do list
        # 如果文件已经存在，可以不需要卸数

        if self.vld_flg == 0:  # 无效
            print(
                f"task_no={self.task_no},task_nm={self.tsk_nm} is invalid, no need_execute"
            )
            return False

        real_date = self.biz_dt + timedelta(
            days=1
        )  # 每月1号供31号的数据，这里的 1 号就是real_date ,31 号就是 biz_dt

        if self.prd == 0:  # 按天跑
            return True
        elif self.prd == 1:  # 按周
            week = real_date.strftime("%w")
            if int(week) == self.prd_dt:
                return True
            else:
                print("run every week")
                return False
        elif self.prd == 2:  # 按旬
            if real_date.day % 10 == self.prd_dt:
                return True
            else:
                print("run every ten days")
                return False
        elif self.prd == 3:  # 按月
            if real_date.day == self.prd_dt:
                return True
            else:
                print("run every month")
                return False
        elif self.prd == 4:  # 按季度
            if self.biz_dt.month in [3, 6, 9, 12] and real_date.day == self.prd_dt:
                return True
            else:
                print("run every season")
                return False
        elif self.prd == 5:  # 按半年
            if self.biz_dt.month in [6, 12] and real_date.day == self.prd_dt:
                return True
            else:
                print("run every half year")
                return False
        else:  # 按年
            if self.biz_dt.month == 12 and real_date.day == self.prd_dt:
                return True
            else:
                print("run every year")
                return False

    @timeit(logger)
    def execute(self):
        # 判断是否跑批
        if not self.need_execute():
            return
        self.record_tsk_begin(self.task_no)
        self.set_dbcli(
            type=self.dtbs_tp,
            host=self.db_ip_addr,
            port=self.pt_no,
            user=self.db_usr_nm,
            pwd=self.db_pwd,
            database=self.dtbs_nm,
        )
        self.export()
        if self.upl_flg:
            self.transfer_cli = self.get_transfer_cli(
                self.prvd_data_mod, self.ip_addr, self.usr_nm, self.psw, self.cd_mod
            )
            logger.info(f"using {self.prvd_data_mod}")
            self.upload(self.data_file.as_posix(), self.remote_data_file.as_posix())
            if self.upload_ok_file == 1:
                self.upload(self.ok_file.as_posix(), self.remote_ok_file.as_posix())
            if self.isLast:
                self.touch_finish_flag()
                self.upload(
                    self.finish_flag.as_posix(), self.remote_finish_flag.as_posix()
                )
        self.record_tsk_end(self.task_no, self.data_num)


@email_on_exception(["zhengzheng@szrcb.com"])
def main():
    isLast = False  # 如果是最后一个文件，那么上传 finish.flag
    task_date = (datetime.today() - timedelta(days=1)).date()  # 如果不传入日期参数，默认昨天
    task_no = None
    if len(sys.argv) not in [2, 3, 4]:
        logger.info("invalid parameters. exit -1")
        logger.info("Usage python supply.py {task_no} [task_date :default yesterday]")
        exit(-1)
    elif len(sys.argv) == 2:
        task_no = sys.argv[1]
    else:

        if sys.argv[2].lower() == "last":
            isLast = True
        elif len(sys.argv[2]) == 10:
            task_date = datetime.strptime(sys.argv[2], "%Y-%m-%d").date()
        elif len(sys.argv[2]) == 8:
            task_date = datetime.strptime(sys.argv[2], "%Y%m%d").date()
        else:
            logger.info("invalid parameters. exit -1")
            exit(-1)
        task_no = sys.argv[1]

    isLast = True if sys.argv[-1].lower() == "last" else False
    ret_code = 0
    if task_no:
        st = SupplyTask(int(task_no), task_date, isLast)
        st.execute()


if __name__ == "__main__":
    main()

